package com.openness.spider;

import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeSet;

import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.openness.spider.commons.Statics;
import com.openness.spider.fetcher.PageFetcher;
import com.openness.spider.hbase.HBaseTablePool;
import com.openness.spider.robotstxt.RobotstxtConfig;
import com.openness.spider.robotstxt.RobotstxtServer;
import com.openness.spider.zk.ZKUtil;

/**
 * @author Rain
 * @version 2013-6-20 下午2:16:47
 */

public class Spider {

	private static final Logger LOGGER = LoggerFactory.getLogger(Spider.class);

	private SpiderConfig config;

	private Set<String> seeds = new HashSet<String>();

	private ZooKeeper zk;

	private String spider;

	private boolean leader;

	private RobotstxtConfig robotstxtConfig = new RobotstxtConfig();

	private PageFetcher pageFetcher;

	private RobotstxtServer robotstxtServer;

	private HtmlBuffer htmlBuffer;

	private UrlBuffer urlBuffer;

	public Spider(SpiderConfig config) {
		this.config = config;

		pageFetcher = new PageFetcher(this.config);

		robotstxtServer = new RobotstxtServer(robotstxtConfig, pageFetcher);
	}

	public void addSeed(String seed) {
		if (seed != null && seed.length() > 0) {
			seeds.add(seed);
		}
	}

	public void removeSeed(String seed) {
		if (seed != null && seed.length() > 0) {
			seeds.remove(seed);
		}
	}

	public String[] getSeeds() {
		return seeds.toArray(new String[seeds.size()]);
	}

	private class MonitorThread extends Thread {

		@Override
		public void run() {
			while (true) {
				try {
					Thread.sleep(config.getMonitorThreadSleepTime());
				} catch (InterruptedException e) {
				}

				List<String> child = null;

				try {
					child = ZKUtil.getChildren(zk, Statics.ZK_THREADS_NODE,
							false);
				} catch (Exception e) {
					LOGGER.error("monitor thread get child error: "
							+ e.toString());

					break;
				}

				if (child != null && child.size() == 0) {
					LOGGER.info("no spider thread exists, request buffer stop");

					htmlBuffer.requestStop();

					urlBuffer.requestStop();

					break;
				}
			}

			LOGGER.info("monitor thread stoped");
		}

	}

	private void leaderJudge() throws Exception {
		List<String> child = ZKUtil.getChildren(zk, Statics.ZK_SPIDERS_NODE,
				false);

		NavigableSet<String> orderedChild = new TreeSet<String>(
				new Comparator<String>() {

					@Override
					public int compare(String str1, String str2) {
						int begin1 = str1.lastIndexOf('_') + 1;
						int begin2 = str2.lastIndexOf('_') + 1;

						long seq1 = Long.valueOf(str1.substring(begin1));
						long seq2 = Long.valueOf(str2.substring(begin2));

						return seq1 < seq2 ? -1 : 1;
					}

				});

		for (String children : child) {
			orderedChild.add(children);
		}

		String lowest = Statics.ZK_SPIDERS_NODE + Statics.ZK_SEPERATOR
				+ orderedChild.first();

		if (lowest.equals(spider)) {
			leader = true;
		} else {
			leader = false;
		}
	}
	
	private void waitSpiderStoped() throws Exception{
		while(true){
			List<String> child = ZKUtil.getChildren(zk, Statics.ZK_SPIDERS_NODE, false);
			
			if(child.size() == 0){
				break;
			}
			
			try {
				Thread.sleep(10000);
			} catch (Exception e) {
			}
		}
	}

	public void start() {
		try {
			HBaseTablePool.initialize(config);

			zk = new ZooKeeper(config.getHbaseZKQuorum(),
					config.getZkSessionTimeout(), null);

			/*
			 * 创建ZK中的基本结点
			 */
			if (!ZKUtil.exists(zk, Statics.ZK_SPIDER, false)) {
				ZKUtil.create(zk, Statics.ZK_SPIDER, Statics.EMPTY_BYTEARRAY,
						Statics.ACL, Statics.CREATEMODE_PERSISTENT);
			}

			if (!ZKUtil.exists(zk, Statics.ZK_SPIDERS_NODE, false)) {
				ZKUtil.create(zk, Statics.ZK_SPIDERS_NODE,
						Statics.EMPTY_BYTEARRAY, Statics.ACL,
						Statics.CREATEMODE_PERSISTENT);
			}

			if (!ZKUtil.exists(zk, Statics.ZK_LOCK_NODE, false)) {
				ZKUtil.create(zk, Statics.ZK_LOCK_NODE,
						Statics.EMPTY_BYTEARRAY, Statics.ACL,
						Statics.CREATEMODE_PERSISTENT);
			}

			if (!ZKUtil.exists(zk, Statics.ZK_THREADS_NODE, false)) {
				ZKUtil.create(zk, Statics.ZK_THREADS_NODE,
						Statics.EMPTY_BYTEARRAY, Statics.ACL,
						Statics.CREATEMODE_PERSISTENT);
			}

			spider = ZKUtil.create(zk, Statics.ZK_SPIDERS_NODE_SUFFIX,
					Statics.EMPTY_BYTEARRAY, Statics.ACL,
					Statics.CREATEMODE_EPHEMERAL_SEQUENTIAL);

			leaderJudge();

			htmlBuffer = new HtmlBuffer(config);

			urlBuffer = new UrlBuffer(config, zk);

			Thread htmlBufferThread = new Thread(htmlBuffer);

			htmlBufferThread.start();

			Thread urlBufferThread = new Thread(urlBuffer);

			urlBufferThread.start();

			for (int index = 0; index < config.getThreads(); index++) {
				new SpiderThread(config, zk, pageFetcher, robotstxtServer,
						htmlBuffer, urlBuffer).start();
			}

			MonitorThread monitor = new MonitorThread();

			monitor.start();

			monitor.join();

			htmlBufferThread.join();

			urlBufferThread.join();

			pageFetcher.shutDown();

			if (ZKUtil.exists(zk, spider, false)) {
				ZKUtil.delete(zk, spider, -1);
			}

			if (leader) {
				waitSpiderStoped();
				
				if(ZKUtil.exists(zk, Statics.ZK_SPIDERS_NODE, false)){
					ZKUtil.delete(zk, Statics.ZK_SPIDERS_NODE, -1);
				}
				
				if (ZKUtil.exists(zk, Statics.ZK_THREADS_NODE, false)) {
					ZKUtil.delete(zk, Statics.ZK_THREADS_NODE, -1);
				}

				if (ZKUtil.exists(zk, Statics.ZK_LOCK_NODE, false)) {
					ZKUtil.delete(zk, Statics.ZK_LOCK_NODE, -1);
				}

				if (ZKUtil.exists(zk, Statics.ZK_SPIDER, false)) {
					ZKUtil.delete(zk, Statics.ZK_SPIDER, -1);
				}
			}
			
			zk.close();
		} catch (Exception e) {
			throw new ExceptionInInitializerError("spider start error: "
					+ e.toString());
		} finally {
		}
	}

	public void stop() {

	}

}
